package org.example.hadoop.service;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.example.hadoop.driver.DefaultClient;
import org.example.hadoop.driver.WordFrequencyMapReduce;
import org.example.hadoop.service.wordFrequency.WordFrequencyMapper;
import org.example.hadoop.service.wordFrequency.WordFrequencyReducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.IOException;

/**
 * @author wwy
 * @date 2020-11-08 14:41
 */
@Service
public class WordFrequencyService {
    private static final Logger LOGGER = LoggerFactory.getLogger(WordFrequencyService.class);
    @Value("${hadoop.inputPath}")
    private String inputPath;
    @Value("${hadoop.outputPath}")
    private String outputPath;
    /**
     * 年份书籍数量
     * 获取目录位置
     * 调用hadoop
     */
    public void yearBookNum(){
        try {
            Configuration conf = new Configuration();
            Job job = Job.getInstance(conf,"yearBookNum");
            job.setMapperClass(WordFrequencyMapper.class);
            job.setReducerClass(WordFrequencyReducer.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
            FileInputFormat.setInputPaths(job, new Path(inputPath));
            FileOutputFormat.setOutputPath(job, new Path(outputPath+"\\yearBookNum"));
            //启动Job任务 ：向 yarn 集群提交这个 job
            boolean res = job.waitForCompletion(true);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }


    }


}
